package ApplicationTest.Example.KafKa

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, TaskContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, KafkaUtils, OffsetRange}

import scala.util.Random

object CreateDirecty {

  private val conf = new SparkConf().setMaster("local[*]").setAppName("Scala Spark Test Application")
  var streamingContext : StreamingContext = new StreamingContext(conf, Seconds(1))

  private val users = Array(
    "4A4D769EB9679C054DE81B973ED5D768", "8dfeb5aaafc027d89349ac9a20b3930f",
    "011BBF43B89BFBF266C865DF0397AA71", "f2a8474bf7bd94f0aabbd4cdd2c06dcf",
    "068b746ed4620d25e26055a9f804385f", "97edfc08311c70143401745a03a50706",
    "d7f141563005d1b5d0d3dd30138f3f62", "c8ee90aade1671a21336c721512b817a",
    "6b67c8c700427dee7552f81f3228c927", "a95f22eabc4fd4b580c011a3161a9d9d")

  private val random = new Random

  private var pointer = -1

  def main(args: Array[String]): Unit = {


    def getUserID : String = {
      pointer += 1
      if (pointer >= users.length){
        pointer = 0
        users(pointer)
      }else{
        users(pointer)
      }
    }



    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "example",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )


    val topics = Array("topicA", "topicB").toSet

    /**
      * 关于链接 kafka -- 关于消费者consumer的记录
      */
    val lines = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    /**
      * 关于DStream
      */
    lines.map(record => (record.key, record.value))

    /**
      * 计算偏移量
      */
    lines.foreachRDD { rdd =>
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //信息记录内容转换获取原始信息
      rdd.foreachPartition { iter =>
        val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
        println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
      }
    }

    streamingContext.start()  //设置开始
    streamingContext.awaitTermination() //等待执行

  }
}
